package com.seaboxdata.utils;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

public class TestKafka {

	public static void main(String[] args) {
		System.setProperty("java.security.auth.login.config", "F:/dfjx/keytab/kafka/kafka_jaas.conf");
		System.setProperty("java.security.krb5.conf", "F:/dfjx/keytab/kafka/krb5.conf");

		Properties props = new Properties();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.21.17.40:6667"); //,172.21.17.41:6667,172.21.17.42:6667
		props.put(ConsumerConfig.GROUP_ID_CONFIG, "gpsMessage-szl");
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
		props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "60000");
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
		props.put("security.protocol", "SASL_PLAINTEXT");
		props.put("sasl.kerberos.service.name", "kafka");
		props.put("sasl.mechanism", "GSSAPI");

		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
		System.out.println("连接成功");
		consumer.subscribe(Collections.singletonList("gpsMessage"));
		System.out.println("订阅成功");

		while (true) {
			System.out.println("--- records------");
			ConsumerRecords<String, String> records = consumer.poll(100);
			System.out.println("循环while");
			for (ConsumerRecord<String, String> record : records) {
				String value = record.value();
				System.out.println("value = "+value);
			}
		}

	}
}
